package com.lzl.consumer;

import com.lzl.common.GateWayPo;
import com.lzl.consumer.models.BlockingQueueMessage;
import com.lzl.consumer.task.ConsumerTask;
import com.lzl.datas.dao.DbDao;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

@Component
@Slf4j
public class Consumer {

    AtomicBoolean running = new AtomicBoolean(true);

    ExecutorService httpExcutor = new ThreadPoolExecutor(3, 5,
            2L, TimeUnit.SECONDS,
            new ArrayBlockingQueue<Runnable>(30));

    @Autowired
    DbDao dbDao;

    public void start() {
        new ReadDbThread().start();
        new ExecuteThread().start();
    }

    class ReadDbThread extends Thread {

        public void run() {
            Long lastId = 0L;
            int i = 0;
            while (running.get()) {
                try {
                    List<GateWayPo> datas = dbDao.getRequestList(lastId);
                    log.info("查询到数据：" + datas);
                    for (GateWayPo po :
                            datas) {
                        BlockingQueueMessage.arrayqueue.put(po);
                        lastId = po.getId();
                    }
                    TimeUnit.SECONDS.sleep(5);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }
        }

    }

    public boolean summit(ConsumerTask task){
        try {
            httpExcutor.submit(task);
            return true;
        } catch (Throwable e) {
            return false;
        }
    }

    class ExecuteThread extends Thread {

        public void run() {
            int i = 0;
            while (running.get()) {
                try {
                    log.info("开始第" + (++i) + "次消费");
                    GateWayPo po = BlockingQueueMessage.arrayqueue.take();
                    if (po != null) {
                        System.out.println("消费" + po.toString());
                        ConsumerTask task = new ConsumerTask(po, dbDao);
                        httpExcutor.submit(task);
                    }

                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }
        }

    }
}
